package cn.xinfei.xdecision.engine.runner.api;

import cn.xinfei.xdecision.common.datamodel.DecisionResult;
import cn.xinfei.xdecision.common.utils.util.DateUtils;
import cn.xinfei.xdecision.common.utils.util.IpUtil;
import cn.xinfei.xdecision.common.utils.util.JsonUtil;
import cn.xinfei.xdecision.constant.EngineResultEnum;
import cn.xinfei.xdecision.constant.GlobalConstant;
import cn.xinfei.xdecision.engine.runner.context.BranchTypeEnum;
import cn.xinfei.xdecision.engine.runner.metadata.PrometheusTagsConstants;
import cn.xinfei.xdecision.engine.runner.service.DecisionResultService;
import cn.xinfei.xdecision.engine.runner.service.EngineReentryService;
import cn.xinfei.xdecision.engine.runner.vo.EngineApiResponse;
import cn.xinfei.xdecision.model.DecisionReentryApiRequest;
import cn.xinfei.xdecision.model.DecisionReentryResponse;
import cn.xinfei.xdecision.redis.RedisLock;
import com.alibaba.fastjson.JSONArray;
import com.dianping.cat.Cat;
import com.github.pagehelper.PageInfo;
import io.micrometer.core.instrument.Metrics;
import io.micrometer.core.instrument.Tags;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;


@Slf4j
@Service
@RefreshScope
public class DecisionReentryApi2 {

    @Autowired
    private EngineReentryService engineReentryService;

    @Resource(name = "reentryThreadPool")
    private ExecutorService reentryThreadPool;


    @Autowired
    private DecisionResultService decisionResultService;
    private static final List<String> DEFAULT_STATES = Arrays.asList(

            //异步申请的
            EngineResultEnum.APPLY.getCode(),
            EngineResultEnum.PROCESSING.getCode(),

            //人行的
            EngineResultEnum.PENDING.getCode(),
            EngineResultEnum.PENDING_REENTRY.getCode(),
            EngineResultEnum.CREDIT_PENDING_APPLY.getCode(),
            EngineResultEnum.CREDIT_PENDING_FETCH.getCode(),
            EngineResultEnum.CREDIT_PENDING_END.getCode(),
            EngineResultEnum.CREDIT_PENDING_TIMEOUT.getCode(),

            //异常的,不再重试
//            EngineResultEnum.EXCEPTION_END.getCode(),
//            EngineResultEnum.EXCEPTION_NO_NEXT_END.getCode(),
//            EngineResultEnum.EXCEPTION_NODE_NULL_END.getCode(),
//            EngineResultEnum.EXCEPTION_TIMEOUT.getCode(),


//            EngineResultEnum.NO_EXISTS.getCode(),//手工补跑
//            EngineResultEnum.NO_DATA_STACK.getCode(),

            //数据源超时异常的
            EngineResultEnum.DATASOURCE_TIMEOUT.getCode(),
            EngineResultEnum.DATASOURCE_EXCEPTION.getCode(),
            EngineResultEnum.DATASOURCE_FAIL.getCode(),
            EngineResultEnum.DATASOURCE_TIMEOUT.getCode());

    private static final String PENDING_REENTRY = "LOCK_PENDING_REENTRY_";

    private static final ConcurrentHashMap<String, Long> PROCESSING_MAIN_DECISION_IDS = new ConcurrentHashMap();

    /**
     * 重入pending的redis锁
     */
    @Value("${xdecision.reentry.timeout}")
    private volatile int reentryTimeout = 60;

    public DecisionReentryResponse reentry(DecisionReentryApiRequest apiRequest) {
        long start = System.currentTimeMillis();
        DecisionReentryResponse decisionApiResponse = new DecisionReentryResponse();
        try {
            String bizId = apiRequest.getRequestId();
            if (StringUtils.isNotEmpty(bizId)) {
                MDC.put(GlobalConstant.TRACE_ID, bizId);
            }
            String decisionId1 = apiRequest.getDecisionId();
            if (StringUtils.isNotEmpty(decisionId1)) {
                MDC.put(GlobalConstant.PID, decisionId1);
            }
            //pending中的main分支记录
            List<DecisionResult> resultPendingList = getRequestId4Reentry(apiRequest, PROCESSING_MAIN_DECISION_IDS);
            if (null == resultPendingList || CollectionUtils.isEmpty(resultPendingList)) {
                decisionApiResponse.setMsg("没有挂起中的决策流");
                log.info("没有挂起中的决策流 apiRequest={}", JsonUtil.toJson(apiRequest));
                return decisionApiResponse;
            }

            List<EngineApiResponse> responses = new ArrayList<>();
            int size = resultPendingList.size();
            decisionApiResponse.setTotal(size);
            for (int i = 0; i < size; i++) {
                DecisionResult decisionResult = resultPendingList.get(i);
                try {
                    reentryThreadPool.submit(() -> {
                        long reentryStart = System.currentTimeMillis();
                        String mainDecisionId = decisionResult.getMainDecisionId();
                        String lockKey = PENDING_REENTRY + mainDecisionId;
                        String ipAddress = IpUtil.getIpAddress();
                        Boolean expire = RedisLock.tryLock(ipAddress, lockKey, reentryTimeout);
                        if (expire) {
                            try {
                                Long id = decisionResult.getId();
                                PROCESSING_MAIN_DECISION_IDS.put(mainDecisionId, id);
                                //再次判断状态是否已经是终态，如果是终态则不处理
                                //pending中的最后一个异常记录
                                List<DecisionResult> preProcessingMainDecisionId = new ArrayList<>();
                                DecisionResult result = new DecisionResult();
                                result.setMainDecisionId(mainDecisionId);
                                preProcessingMainDecisionId.add(result);
                                List<DecisionResult> confirmProcessingMainDecisionId = findLastPendingDecisionResult(preProcessingMainDecisionId, apiRequest);
                                if (CollectionUtils.isNotEmpty(confirmProcessingMainDecisionId)) {
                                    DecisionResult confirmDecisionResult = confirmProcessingMainDecisionId.get(0);
                                    String code = confirmDecisionResult.getCode();
                                    if (DEFAULT_STATES.contains(code) || (StringUtils.isNotEmpty(apiRequest.getCode()) && apiRequest.getCode().equals(code))) {
                                        String requestId = confirmDecisionResult.getRequestId();
                                        String decisionId = confirmDecisionResult.getDecisionId();
                                        MDC.put(GlobalConstant.TRACE_ID, requestId);
                                        MDC.put(GlobalConstant.PID, decisionId);
                                        String engineCode = confirmDecisionResult.getEngineCode();
                                        List<EngineApiResponse> engineApiResponses = engineReentryService.reentry(requestId, decisionId, engineCode, "XXL", ipAddress, lockKey);
                                        responses.addAll(engineApiResponses);
                                    } else {
                                        //暂时info,稳定后改未debug
                                        log.info("记录已经是终态，不需要再次重入");
                                    }
                                } else {
                                    log.info("不存在待处理的记录,mainDecisionId={}", mainDecisionId);
                                }
                            } catch (Exception e) {
                                log.error("重入接口处理异常，mainDecisionId={},msg={}", e.getMessage(), mainDecisionId, e);
                                Cat.logError(e);
                                Metrics.counter("xdecision_reentry_total", Tags.of("result", "exception")).increment();
                            } finally {
                                long reentryEnd = System.currentTimeMillis();
                                long reentryDuration = reentryEnd - reentryStart;
                                Metrics.gauge("xdecision_reentry_respond", PrometheusTagsConstants.XENGINE_API_RESPOND_TAGS, reentryDuration);
                                log.info("reentryDuration={}", reentryDuration);
                                PROCESSING_MAIN_DECISION_IDS.remove(mainDecisionId);
                                RedisLock.releaseLock(ipAddress, lockKey);
                                MDC.remove(GlobalConstant.NID);
                                MDC.remove(GlobalConstant.PID);
                                MDC.remove(GlobalConstant.TRACE_ID);
                                MDC.clear();
                            }
                        } else {
                            //暂时info,稳定后改未debug
                            log.info("申请分布式锁失败，clientId={},lockKey={}", ipAddress, lockKey);
                        }
                    });
                } catch (Exception e) {
                    log.error("reentry线程池添加任务失败,msg={}", e.getMessage(), e);
                }
            }
            //只是说明任务执行成功，不代表业务含义，只说明没有异常
            decisionApiResponse.setCode("000000");
            decisionApiResponse.setResult(JSONArray.toJSONString(responses));
            long respond = System.currentTimeMillis() - start;
            log.info("reentry respond={},decisionApiResponse={}", respond, decisionApiResponse);
        } catch (Exception e) {
            log.error("重入接口异常，error={},apiRequest={}", e.getMessage(), JsonUtil.toJson(apiRequest), e);
            decisionApiResponse.setMsg("重入接口异常:" + e.getMessage());
            Cat.logError(e);

            Metrics.counter("xdecision_reentry_total", Tags.of("result", "exception")).increment();
        }


        return decisionApiResponse;
    }

    private List<DecisionResult> findLastPendingDecisionResult(List<DecisionResult> pendingRequestIds, DecisionReentryApiRequest apiRequest) {
        int size = pendingRequestIds.size();

        Map<String, DecisionResult> decisionIdMap = new HashMap<>();
        Map<String, String> parentDecisionIdMap = new HashMap<>();
        for (int i = 0; i < size; i++) {
            DecisionResult requestIdResult = pendingRequestIds.get(i);
            String mainDecisionId = requestIdResult.getMainDecisionId();
            List<DecisionResult> decisionResultNew = decisionResultService.getDecisionResultByMainDecisionId(mainDecisionId);
            for (int i1 = 0; i1 < decisionResultNew.size(); i1++) {
                //查询当前requestId下的左右决策记录,从当前记录列表中，找出最后一个pending的记录
                DecisionResult decisionResult = decisionResultNew.get(i1);
                String code = decisionResult.getCode();
                String decisionId = decisionResult.getDecisionId();
                if (DEFAULT_STATES.contains(code) || (StringUtils.isNotEmpty(apiRequest.getCode()) && apiRequest.getCode().equals(code))) {
                    String branch = decisionResult.getBranch();
                    if (branch.equals(BranchTypeEnum.MAIN.name()) || branch.equals(BranchTypeEnum.CHAMPION.name()) || branch.equals(BranchTypeEnum.CHILD.name())) {
                        String parentDecisionId = decisionResult.getParentDecisionId();
                        decisionIdMap.put(decisionId, decisionResult);
                        if (StringUtils.isNotEmpty(parentDecisionId)) {
                            parentDecisionIdMap.put(parentDecisionId, parentDecisionId);
                        }
                    } else if (branch.equals(BranchTypeEnum.CHALLENGE.name())) {
                        log.info("{}挑战者记录暂不进行重入", decisionId);
                    } else {
                        log.info("{}记录暂不进行重入,branch={}", decisionId, branch);
                    }
                } else {
                    log.info("{}记录暂不进行重入,code={},branch={}", decisionId, code, decisionResult.getBranch());
                }
            }
        }
        List<DecisionResult> lastPendingResults = new ArrayList<>();
        Set<Map.Entry<String, DecisionResult>> entries = decisionIdMap.entrySet();
        Iterator<Map.Entry<String, DecisionResult>> iterator = entries.iterator();
        while (iterator.hasNext()) {
            Map.Entry<String, DecisionResult> next = iterator.next();
            String decisionId = next.getKey();
            if (!parentDecisionIdMap.containsKey(decisionId)) {
                //最后一个pending的决策流
                DecisionResult lastPendingResult = next.getValue();
                lastPendingResults.add(lastPendingResult);
                log.info("pending的决策流,decisionId={},branch={}", lastPendingResult.getDecisionId(), lastPendingResult.getBranch());
//                break;
            }
        }
        return lastPendingResults;
    }

    /**
     * 查询pending中的决策请求快照
     *
     * @param apiRequest
     * @return
     */
    private List<DecisionResult> getRequestId4Reentry(DecisionReentryApiRequest apiRequest, ConcurrentHashMap<String, Long> processingIds) {
        List<String> states = new ArrayList<>();
        Date startTime = apiRequest.getStart();
        Date endTime = apiRequest.getEnd();
        if (null == apiRequest) {
            apiRequest = new DecisionReentryApiRequest();
        }
        //如果传入了state，则优先使用state状态
        if (StringUtils.isNotEmpty(apiRequest.getCode())) {
            states = Arrays.asList(apiRequest.getCode());
        } else {
            states = DEFAULT_STATES;
        }

        if (null == startTime) {
            startTime = DateUtils.getBeforeNumDay(1);
        }
        if (null == endTime) {
            endTime = new Date();
        }
        PageInfo<DecisionResult> decisionResultPending = decisionResultService.getMainRequestIdsByReentryRequest(apiRequest, states, startTime, endTime, processingIds);
        if (null == decisionResultPending) {
//            log.info("当前条件下，没有pending中的记录");
            return null;
        }
        return decisionResultPending.getList();
    }

}
